package com.arch.flink.biz;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.KeyedStateStore;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author pizhihui
 * @date 2020-12-13
 */
public class ProcessWindowDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> dataStream = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
                String[] fields = line.split(",");
                for (String word : fields) {
                    out.collect(new Tuple2<>(word, 1));
                }
            }
        })
                .keyBy(x -> x.f0)
                .timeWindow(Time.seconds(10), Time.seconds(5))
                .process(new MyProcessWindow());

        result.print().setParallelism(1);

        env.execute("TimeWindowWordCount");

    }


    public static class MyProcessWindow
            extends ProcessWindowFunction<Tuple2<String, Integer>,
            Tuple2<String, Integer>,
            String,
            TimeWindow> {

        FastDateFormat instance = FastDateFormat.getInstance("HH:mm:ss");

        @Override
        public void process(String s,
                            Context context,
                            Iterable<Tuple2<String, Integer>> elements,
                            Collector<Tuple2<String, Integer>> out) throws Exception {

            long l = context.currentProcessingTime();
            long start = context.window().getStart();
            long end = context.window().getEnd();

            KeyedStateStore keyedStateStore = context.windowState();
            long l1 = context.currentWatermark();

            System.out.println("当天系统时间: " + instance.format(System.currentTimeMillis()));
            System.out.println("window的处理时间: " + instance.format(l));
            System.out.println("window的开始时间: " + instance.format(start));
            System.out.println("window的结束时间: " + instance.format(end));

            int sum = 0;
            for (Tuple2<String, Integer> element : elements) {
                sum += 1;
            }

            out.collect(Tuple2.of(s, sum));

        }
    }
}
